package com.li.flink.datastream

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

object Sink2Redis {
  def main(args: Array[String]): Unit = {
    val conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build()
    val inputPath = "/Users/lijiacen/Downloads/test.txt"
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.readTextFile(inputPath).map(m => {
      var arr = m.split(" ")
      (arr(0), arr(1))
    }).addSink(new RedisSink[(String, String)](conf, new RedisExampleMapper))

    env.execute("Sink2Redis")
  }

  class RedisExampleMapper extends RedisMapper[(String, String)] {
    override def getCommandDescription: RedisCommandDescription = {
      new RedisCommandDescription(RedisCommand.HSET, "sink2redis")
    }

    override def getKeyFromData(data: (String, String)): String = data._1

    override def getValueFromData(data: (String, String)): String = data._2
  }
}
